package com.caissa.springboot.starter.es.autoconfigure;


import com.base.springboot.starter.es.config.ElasticsearchConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.http.HttpHeaders;

import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLContext;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 *
 * @author  dsk
 * @since  2020/06/08
 */
@Configuration
@EnableConfigurationProperties(ElasticsearchConfig.class)
@Slf4j
public class ElasticsearchConfiguration {
    @Autowired
    private ElasticsearchConfig elasticsearchConfig;
    @Bean
    @ConditionalOnClass(RestHighLevelClient.class)
    public RestHighLevelClient client(){
        String[] address = elasticsearchConfig.getEndpoints().replace("[","").replace("]","").split(",");
        log.info("ElasticsearchConfig init :",elasticsearchConfig.getEndpoints());
        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo(address)
                .withBasicAuth(elasticsearchConfig.getUsername(),elasticsearchConfig.getPassword())
                .withConnectTimeout(elasticsearchConfig.getConnectionTimeout())
                .withSocketTimeout(elasticsearchConfig.getSocketTimeout())
                .build();
        HttpHost[] httpHosts = formattedHosts(clientConfiguration.getEndpoints(), clientConfiguration.useSsl()).stream().map(HttpHost::create).toArray(x -> new HttpHost[x]);
        RestClientBuilder builder = RestClient.builder(httpHosts);
        if (clientConfiguration.getPathPrefix() != null) {
            builder.setPathPrefix(clientConfiguration.getPathPrefix());
        }

        HttpHeaders headers = clientConfiguration.getDefaultHeaders();
        if (!headers.isEmpty()) {
            Header[] httpHeaders = headers.toSingleValueMap().entrySet().stream().map(it -> new BasicHeader(it.getKey(), it.getValue())).toArray(x -> new Header[x]);
            builder.setDefaultHeaders(httpHeaders);
        }

        builder.setHttpClientConfigCallback(clientBuilder -> {
            Optional<SSLContext> sslContext = clientConfiguration.getSslContext();
            Optional<HostnameVerifier> hostNameVerifier = clientConfiguration.getHostNameVerifier();
            sslContext.ifPresent(clientBuilder::setSSLContext);
            hostNameVerifier.ifPresent(clientBuilder::setSSLHostnameVerifier);

            Duration connectTimeout = clientConfiguration.getConnectTimeout();
            Duration timeout = clientConfiguration.getSocketTimeout();
            RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
            if (!connectTimeout.isNegative()) {
                requestConfigBuilder.setConnectTimeout(Math.toIntExact(connectTimeout.toMillis()));
                requestConfigBuilder.setConnectionRequestTimeout(Math.toIntExact(connectTimeout.toMillis()));
            }

            if (!timeout.isNegative()) {
                requestConfigBuilder.setSocketTimeout(Math.toIntExact(timeout.toMillis()));
            }

            clientBuilder.setDefaultRequestConfig(requestConfigBuilder.build());
            clientConfiguration.getProxy().map(HttpHost::create).ifPresent(clientBuilder::setProxy);
            return clientBuilder;
        });
        return new RestHighLevelClient(builder);
    }
    private static List<String> formattedHosts(List<InetSocketAddress> hosts, boolean useSsl) {
        return hosts.stream().map(it -> (useSsl ? "https" : "http") + "://" + it).collect(Collectors.toList());
    }
}

